/**
 * Copyright(c) 2015 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-8-1 下午12:44 Created by Daisy
 */
package com.jade.framework.mom.rocketmq;

import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.jade.framework.mom.*;

/**
 * RocketMQ的生产者。
 *
 * @author Daisy
 * @version 1.0 14-8-1 下午12:44
 */
public class RocketmqProducer extends AbstractProducer
{
    DefaultMQProducer producer;

    public void setProducer(DefaultMQProducer producer)
    {
        this.producer = producer;
    }

    @Override
    protected void doInit()
    {
        //do nothing
    }

    @Override
    protected void doDestroy()
    {
        //do nothing
    }

    @Override
    protected void doStart()
            throws Exception
    {
        producer.start();
    }

    @Override
    protected void doStop()
            throws Exception
    {
        producer.shutdown();
    }

    public MomSendResult send(MomMessage message)
            throws MomException
    {
        if (!isStarted()) {
            throw new MomException(MomConstants.ERR_PRODUCER_NOT_STARTED, "Producer not started");
        }

        Message rmqMsg = RocketmqUtils.toRmqMessage(message);
        try {
            SendResult result = producer.send(rmqMsg);
            message.setId(result.getMsgId());
            return RocketmqUtils.toMomResult(result, message);
        }
        catch (Throwable t) {
            throw RocketmqUtils.handleException(t);
        }
    }

    public void send(MomMessage message, MomSendCallback callback)
            throws MomException
    {
        if (!isStarted()) {
            throw new MomException(MomConstants.ERR_PRODUCER_NOT_STARTED, "Producer not started");
        }
        Message rmqMsg = RocketmqUtils.toRmqMessage(message);
        try {
            producer.send(rmqMsg, new SendCallbackAdapter(message, callback));
        }
        catch (Throwable t) {
            throw RocketmqUtils.handleException(t);
        }
    }

    protected class SendCallbackAdapter implements SendCallback
    {
        protected MomSendCallback callback;
        protected MomMessage message;

        SendCallbackAdapter(MomMessage message, MomSendCallback callback)
        {
            this.message = message;
            this.callback = callback;
        }

        public void onSuccess(SendResult sendResult)
        {
            message.setId(sendResult.getMsgId());
            callback.onSuccess(RocketmqUtils.toMomResult(sendResult, message));
        }

        public void onException(Throwable throwable)
        {
            callback.onException(RocketmqUtils.handleException(throwable));
        }
    }
}
